package at.jaki.ubb.crawling.dataTransformer;

import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.Future;

import org.apache.log4j.Logger;

import at.jaki.ubb.crawling.api.DataTransformer;
import at.jaki.ubb.crawling.appThreads.DataTransformerResultManager;
import at.jaki.ubb.crawling.core.Constants;
import at.jaki.ubb.crawling.core.ThreadPool;
import at.jaki.ubb.crawling.model.RawData;
import at.jaki.ubb.crawling.model.config.DataFinderManagerConfig;
import at.jaki.ubb.crawling.model.config.DataTransformerConfig;
import at.jaki.ubb.crawling.model.config.DataTransformerManagerConfig;

/**
 * This class manages all the data transforming operations. used to call all the
 * datatransformers.
 * 
 * @author Attila Jakabfi
 * 
 */
@SuppressWarnings({ "unchecked", "rawtypes" })
public class DataTransformerManager {

	private static final Logger LOG = Logger.getLogger(DataTransformerManager.class);

	private static DataTransformerManager instance;

	/**
	 * Initializes the {@link DataTransformerManager}.
	 * 
	 * @param configObject
	 * @return true if initialization was successful
	 */
	public static boolean init(DataTransformerManagerConfig configObject) {
		instance = new DataTransformerManager(configObject);
		return instance != null;
	}

	public static DataTransformerManager getInstance() {
		if (instance == null) {
			throw new IllegalStateException("DataTransformerManager is not initialized.");
		}
		return instance;
	}

	/**
	 * Map containing all started data transformer threads.
	 */
	private Map<String, Collection<DataTransformerThread>> threadMap = new HashMap<String, Collection<DataTransformerThread>>();

	/**
	 * @param configObject the {@link DataFinderManagerConfig} object.
	 *        Containing all the data require for initialization.
	 */
	private DataTransformerManager(DataTransformerManagerConfig configObject) {
		for (DataTransformerConfig instance : configObject.getDataTransformers()) {
			DataTransformerThread dft = new DataTransformerThread(instance.getClazzName());
			dft.setName(instance.getLogName());
			if (dft.isInitialized()) {
				Collection<String> typesList = dft.getTypes();
				for (String type : typesList) {
					if (!threadMap.containsKey(type)) {
						threadMap.put(type, new LinkedList<DataTransformerThread>());
					}
					threadMap.get(type).add(dft);
				}
			}
		}
	}

	/**
	 * Will find a {@link DataTransformer} of type <code>type</code> and execute
	 * it for <code>data</code>.
	 * 
	 * @param data target
	 */
	public synchronized <RD> void call(RawData<RD> data) {
		LOG.debug("Called DataTransformerManager for type '" + data.getType() + "' and data '" + data.toString() + "'");
		if (!threadMap.containsKey(data.getType())) {
			LOG.warn("No data transformer registered for type: " + data.getType());
			LOG.debug("Processing of " + data.toString() + " skipped");
			return;
		}
		Collection<DataTransformerThread> dataTransformers = threadMap.get(data.getType());
		outer: while (true) {
			for (DataTransformerThread thread : dataTransformers) {
				if (!thread.isBusy()) {
					thread.setData(data);
					LOG.debug("submitting raw data '" + data + "' to '" + thread.getName() + "'");
					try {
						Future future = ThreadPool.getInstance().submit(thread);
						DataTransformerResultManager.getInstance().addTask(future);
						LOG.debug("'" + thread.getName() + "' has been called");
					} catch (Exception e) {
						LOG.error("Error with data transformer: " + thread.getName());
						LOG.error("Error while executing task. Stack trace:");
						if (LOG.isDebugEnabled()) {
							e.printStackTrace();
						}
					}
					break outer;
				}
			}
			try {
				Thread.sleep(Constants.SLEEP_TIME);
			} catch (InterruptedException e) {
				LOG.warn("sleep interrupted " + e.getMessage());
			}
		}
	}

}
